package com.syf.starter.kafka.handler;

import com.syf.core.engine.SyFrameContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.core.io.support.SpringFactoriesLoader;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Kafka消息处理器管理器
 * 负责加载和管理所有的消息处理器
 */
@Slf4j
public class KafkaMessageHandlerManager {

    private final SyFrameContext syFrameContext;

    // topic -> handlers映射，一个topic可以有多个处理器
    private final Map<String, List<KafkaMessageHandler>> topicHandlers = new ConcurrentHashMap<>();

    // 所有已注册的处理器
    private final List<KafkaMessageHandler> allHandlers = new ArrayList<>();

    public KafkaMessageHandlerManager(SyFrameContext syFrameContext) {
        this.syFrameContext = syFrameContext;
        init();
    }

    private void init() {
        // 使用Spring SPI机制加载所有KafkaMessageHandler实现类
        List<KafkaMessageHandler> handlers = SpringFactoriesLoader.loadFactories(
                KafkaMessageHandler.class,
                KafkaMessageHandlerManager.class.getClassLoader());

        if (handlers.isEmpty()) {
            log.warn("未找到任何Kafka消息处理器");
            return;
        }

        for (KafkaMessageHandler handler : handlers) {
            String handlerName = handler.getHandlerName();

            // 获取处理器要处理的topics
            List<String> topics = handler.getTopics();
            if (topics == null || topics.isEmpty()) {
                log.warn("处理器{}未指定要处理的topics，将被忽略", handlerName);
                continue;
            }

            // 将处理器添加到所有处理器列表中
            allHandlers.add(handler);

            // 将处理器添加到对应topic的处理器列表中
            for (String topic : topics) {
                topicHandlers.computeIfAbsent(topic, k -> new ArrayList<>()).add(handler);
                log.info("已为topic[{}]加载处理器: {}", topic, handlerName);
            }
        }

        log.info("共加载了{}个Kafka消息处理器，处理{}个topics", allHandlers.size(), topicHandlers.size());
    }

    /**
     * 处理Kafka消息
     *
     * @param record Kafka消息记录
     * @return 是否有处理器成功处理了消息
     */
    public boolean handleMessage(ConsumerRecord<String, String> record) {
        String topic = record.topic();
        boolean handled = false;

        // 遍历所有处理器，检查消息的topic是否在处理器的topic集合中
        for (KafkaMessageHandler handler : allHandlers) {
            List<String> handlerTopics = handler.getTopics();
            if (handlerTopics != null && handlerTopics.contains(topic)) {
                try {
                    if (handler.handleMessage(record, syFrameContext)) {
                        handled = true;
                        if (log.isDebugEnabled()) {
                            log.debug("消息已被topic[{}]的处理器{}成功处理", topic, handler.getHandlerName());
                        }
                    }
                } catch (Exception e) {
                    log.error("topic[{}]的处理器{}处理消息时发生错误", topic, handler.getHandlerName(), e);
                }
            }
        }

        return handled;
    }

    /**
     * 获取所有已注册的处理器
     *
     * @return 处理器列表
     */
    public List<KafkaMessageHandler> getHandlers() {
        return allHandlers;
    }


    /**
     *
     * 获取订阅的topic集合
     * */
    public Set<String> getAllTopics() {
        return topicHandlers.keySet();
    }
}